package net.techfuser.sync;

import lombok.extern.slf4j.Slf4j;
import org.apache.flink.configuration.Configuration;
import org.apache.flink.streaming.api.functions.sink.RichSinkFunction;

import java.sql.Connection;
import java.util.Date;


@Slf4j
class OrderDetailMySqlSink extends RichSinkFunction<DMLData> {

    private Connection connection = null;

    @Override
    public void open(Configuration parameters) throws Exception {

        String url = "jdbc:mysql://qnh-flink-data.mysql.rds.aliyuncs.com:3306/flink_data_output?useUnicode=true&characterEncoding=utf-8";
        String user = "qnh_flink";
        String passwd = "Md7fpbWSHpBa";
        String driver = "com.mysql.jdbc.Driver";

        connection = DbUtils.getConnection(driver, url, user, passwd);
    }

    @Override
    public void invoke(DMLData dml, Context context) throws Exception {
        if(DMLType.INSERT == dml.getType()){
            DbUtils.insert(connection, "order_detail", dml.getData());
        }else if(DMLType.UPDATE == dml.getType()){
            dml.getData().put("last_modify_time", new Date());
            DbUtils.excuteUpdate(connection, "order_detail", dml.getData(), dml.getWhere());
        }else if(DMLType.DELETE == dml.getType()) {
            DbUtils.delete(connection, "order_detail", dml.getWhere());
        }else{
            log.warn("未处理的类型");
        }
    }

    @Override
    public void close() throws Exception {
        DbUtils.releaseResource(connection);
    }

}